package com.example.yurpceasy.register;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import com.example.yurpceasy.config.RegistryConfig;
import com.example.yurpceasy.model.ServiceMetaInfo;
import io.etcd.jetcd.*;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

@Slf4j
public class EtcdRegistry implements Registry {

    /**
     * 作为rcp服务要存的键值对的根节点
     */
    private static final String ETCD_ROOT_PATH = "/lu-rpc/";


    /**
     * 本机注册的节点 key 集合 (用于维护续签）
     */
    private final Set<String> localRegisterNodeKeySet = new HashSet<>();

    /**
     * 注册中心服务缓存
     */
    private final RegistryServiceCache registryServiceCache =new RegistryServiceCache();


    /**
     * 正在监听的 key 集合
     */
    private final Set<String> watchingKeySet = new ConcurrentHashSet<>();


    private Client client;

    private KV kvClient;

    @Override
    public void init(RegistryConfig registerConfig) {
        client = Client.builder()
                .endpoints(registerConfig.getAddress())
                .connectTimeout(Duration.ofMillis(registerConfig.getTimeout()))
                .build();
        kvClient = client.getKVClient();
        heartBeat();
    }

    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        // etcd租约客户端
        Lease leaseClient = client.getLeaseClient();
        // 创建一个30s的租约
        long leaseId = leaseClient.grant(30).get().getID();

        // 存键值对
        String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        ByteSequence key = ByteSequence.from(registerKey, StandardCharsets.UTF_8);
        ByteSequence val = ByteSequence.from(JSONUtil.toJsonStr(serviceMetaInfo), StandardCharsets.UTF_8);

        //将键值对与租约关联，并设置过期时间
        PutOption putOption = PutOption.builder().withLeaseId(leaseId).build();
        kvClient.put(key, val, putOption).get();

        //注册成功，将节点信息添加到本地缓存
        localRegisterNodeKeySet.add(registerKey);
    }

    @Override
    public void unRegistry(ServiceMetaInfo serviceMetaInfo) {
        String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        ByteSequence key = ByteSequence.from(registerKey, StandardCharsets.UTF_8);
        try {
            kvClient.delete(key).get();
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }

        //注销成功，将节点信息移除本地缓存
        localRegisterNodeKeySet.add(registerKey);
    }

    /**
     * 发现服务，根据服务名称作为前缀，从Etcd获取服务下的节点列表
     *
     * @param serviceKey
     * @return
     */
    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        // 优先从服务缓存获取服务
        List<ServiceMetaInfo> cacheServiceMetaInfoList = registryServiceCache.readCache();
        if (cacheServiceMetaInfoList!=null){
            return cacheServiceMetaInfoList;
        }

        // etcd规则，前缀搜素结尾要加 ‘/’
        String searchPrefix = ETCD_ROOT_PATH + serviceKey + "/";
        ByteSequence keyPrefix = ByteSequence.from(searchPrefix, StandardCharsets.UTF_8);

        try {
            GetOption getOption = GetOption.builder().isPrefix(true).build();
            List<KeyValue> kvs = kvClient.get(keyPrefix, getOption).get().getKvs();

            List<ServiceMetaInfo> serviceMetaInfos = kvs.stream().map(keyValue -> {
                // 监听key (一般第一次发现时监听)
                String key = keyValue.getKey().toString(StandardCharsets.UTF_8);
                watch(key);

                // 查到的节点服务信息
                String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                return JSONUtil.toBean(value, ServiceMetaInfo.class);
            }).collect(Collectors.toList());

            // 写入服务缓存
            registryServiceCache.writeCache(serviceMetaInfos);
            return serviceMetaInfos;
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException("获取服务列表失败");
        }
    }

    @Override
    public void destroy() {
        log.info("当前节点下线");

        for (String key : localRegisterNodeKeySet) {
            try {
                ByteSequence from = ByteSequence.from(key, StandardCharsets.UTF_8);
                kvClient.delete(from).get();
            } catch (Exception e) {
                throw new RuntimeException(key + "节点下线失败", e);
            }
        }

        if (kvClient != null) {
            kvClient.close();
        }
        if (client != null) {
            client.close();
        }
    }


    // 心跳检测的时间要小于 节点的存活时间
    @Override
    public void heartBeat() {
        CronUtil.schedule("0/10 * * * * *", new Task() {
            @Override
            public void execute() {
                // 遍历所有本地缓存的节点，本地缓存的节点当然都是本地注册节点
                // 续签也只能帮自己提供的服务续签，不能为别人续签，因为也不知道别人是死是活
                for (String key : localRegisterNodeKeySet) {
                    try {
                        ByteSequence from = ByteSequence.from(key, StandardCharsets.UTF_8);
                        List<KeyValue> kvs = kvClient.get(from).get().getKvs();

                        // 没找到，说明节点已经过期，需要重新启动手动注册了
                        if (CollUtil.isEmpty(kvs)) {
                            continue;
                        }
                        // 找到了，说明没过期，再帮它注册一次
                        KeyValue keyValue = kvs.get(0);
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                        register(serviceMetaInfo);
                    } catch (Exception e) {
                        throw new RuntimeException(key + "续签失败", e);
                    }
                }
            }
        });

        CronUtil.setMatchSecond(true); //支持秒级
        CronUtil.start();
    }


    /**
     *  监听etcd key的函数
     *  注意同一个key监听后，被删除了又被注册，不用重新监听，因为本来就还在监听
     * @param serviceNodeKey
     */
    @Override
    public void watch(String serviceNodeKey) {
        Watch watchClient = client.getWatchClient();
        // 之前未被监听的，开启监听
        boolean newWatch = watchingKeySet.add(serviceNodeKey);
        if (newWatch){
            ByteSequence from = ByteSequence.from(serviceNodeKey, StandardCharsets.UTF_8);
            watchClient.watch(from,watchResponse -> {
                for (WatchEvent event : watchResponse.getEvents()) {
                    switch (event.getEventType()){
                        // key 被删除时触法
                        case DELETE :
                            registryServiceCache.clearCache();
                            break;
                        case PUT:
                        default:
                            break;
                    }
                }
            });
        }
    }
}
